package cn.dfun.sample.flink.tabletest

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.scala._
import org.apache.flink.table.api.DataTypes
import org.apache.flink.table.descriptors.{Csv, Kafka, Schema}

// kafka进 kafka出
// kafka只支持append模式
// todo 只输出了一条
// ./bin/kafka-console-producer.sh --broker-list node-01:9092 --topic
//sensor
// ./bin/kafka-console-consumer.sh --bootstrap-server node-01:9092 --topic sinktest
object KafkaPiplineTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val tableEnv = StreamTableEnvironment.create(env)

    tableEnv.connect(new Kafka()
      .version("0.11")
      .topic("sensor")
      .property("zookeeper.connect", "node-01:2181")
      .property("bootstrap.servers", "node-01:9092")
    )
      .withFormat(new Csv())
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("timestamp", DataTypes.BIGINT())
        .field("temperature", DataTypes.DOUBLE())
      )
      .createTemporaryTable("kafkaInputTable")
    val sensorTable = tableEnv.from("kafkaInputTable")

    val resultTable = sensorTable
      .select('id, 'temperature)
      .filter('id === "sensor_1")

    val aggTable = sensorTable
      .groupBy('id) // 基于id分组
      .select('id, 'id.count as 'count)

    // 输出到kafka
    tableEnv.connect(new Kafka()
      .version("0.11")
      .topic("sinktest")
      .property("zookeeper.connect", "node-01:2181")
      .property("bootstrap.servers", "node-01:9092")
    )
      .withFormat(new Csv())
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("temp", DataTypes.DOUBLE())
      )
      .createTemporaryTable("kafkaOutputTable")
    resultTable.insertInto("kafkaOutputTable")
    env.execute("kafka pipeline test")
  }
}
